package com.moon.game.service;

import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.LinkedBlockingQueue;

import org.apache.log4j.Logger;

import com.moon.game.exception.GException;


@SuppressWarnings("rawtypes")
public class SQLQueue {
	private static final Logger log = Logger.getLogger(SQLQueue.class);
	public static SQLQueue inst;
	public static final int dbThreadNum = 8;
	private static ConcurrentHashMap<Integer, LinkedBlockingQueue<NotifyItem>> operaQueues = new ConcurrentHashMap<Integer, LinkedBlockingQueue<NotifyItem>>();

	public synchronized static void init() throws GException {
		if (inst == null) {
			inst = new SQLQueue();
		}
		// 线程数散列
		for (int i = 0; i < dbThreadNum; i++) {
			operaQueues.put(i, new LinkedBlockingQueue<NotifyItem>());
		}
	}

	static class NotifyItem<T> {
		int type;
		int action;
		private Object object;
		public NotifyItem(int type, int action, Object object){
			this.type = type;
			this.action = action;
			this.object = object;
		}
		public int getType() {
			return type;
		}
		public void setType(int type) {
			this.type = type;
		}
		public int getAction() {
			return action;
		}
		public void setAction(int action) {
			this.action = action;
		}
		public Object getObject() {
			return object;
		}
		public void setObject(Object object) {
			this.object = object;
		}		
	}

	/**
	 * 
	 * @param key
	 * @param table
	 * @param action
	 * @param object
	 */
	public void addExecute(String key, int table, int action, Object object) {
		// +1 区分AppServer的散列,使每个线程比较平均
		int index = (int)(Math.abs((key+"1").hashCode()) % dbThreadNum);
		
		NotifyItem item = new NotifyItem(table, action, object);
		LinkedBlockingQueue<NotifyItem> operaQueue = operaQueues.get(index);
		if(operaQueue.size()>500){
			log.error("hash "+index+" execute sql queue size:"+operaQueue.size());
		}	
		operaQueue.add(item);
		
		// 线程卡死，重新启动线程执行
		if(operaQueue.size()>=5000){
			log.error("start new thread "+index);
			// TODO: 下面代码临时关闭，以后需要打开
			/*Thread t = new DBExecuteThread(index);
			t.start();*/
		}
	}
	
	public LinkedBlockingQueue<NotifyItem> getQueue(int index){
		return operaQueues.get(index);
	}
}
